package com.abocode.aichat.messenger.application;


import com.abocode.aichat.core.domain.entity.Message;
import com.abocode.aichat.data.Entity;
import com.abocode.aichat.data.MessageStanza;
import com.abocode.aichat.data.MessageStanzaImpl;
import com.abocode.aichat.server.context.pipeline.MessagePipeline;
import com.abocode.aichat.server.context.delivery.reply.Replies;
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * @author guanxf
 */
@Service("messageGatewayPipeline")
public class RedisMessagePipelineImpl implements MessagePipeline {
    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void push(Entity entity, MessageStanza stanza) {
        String offlineKey = RedisKeyHolder.CHAT_MESSAGE_PREFIX + entity.getNode();
        redisTemplate.persist(offlineKey);
        BoundHashOperations<String,String,String> boundHashOperations = redisTemplate.boundHashOps(offlineKey);
        boundHashOperations.put(stanza.id(),JSON.toJSONString(stanza));
    }

    public Set<MessageStanza> get(Entity entity) {
        String offlineKey =  RedisKeyHolder.CHAT_MESSAGE_PREFIX + entity.getNode();
        if (redisTemplate.hasKey(offlineKey)) {
            BoundHashOperations<String,String,String> boundHashOperations = redisTemplate.boundHashOps(offlineKey);
            List<String> messages=boundHashOperations.values();
            Set<MessageStanza> stanzas=null;
            if(messages!=null && !messages.isEmpty()){
                stanzas = new HashSet<>();
                for(String message:messages){
                    Message msg = JSON.parseObject(message, Message.class);
                    MessageStanzaImpl stanza=new MessageStanzaImpl();
                    stanza.setBody(msg);
                    stanzas.add(stanza);
                }
            }
            redisTemplate.delete(offlineKey);
            return stanzas;
        }
        return null;
    }

    @Override
    public void sendMessageToChannel(String mes, String channel){
        redisTemplate.convertAndSend(channel,mes);
    }

    @Override
    public void confirm(Entity entity, Object msgId) {
        String offlineKey =  RedisKeyHolder.CHAT_MESSAGE_PREFIX + entity.getNode();
        if (redisTemplate.hasKey(offlineKey)) {
            BoundHashOperations<String,String,String> boundHashOperations = redisTemplate.boundHashOps(offlineKey);
            boundHashOperations.delete(msgId);
        }
    }

    @Override
    public Replies messages(Channel channel, Entity toEntity) {
        Set<MessageStanza> offlines = get(toEntity);
        Replies replies = null;
        if (offlines != null && !offlines.isEmpty()) {
            replies = new Replies();
            for (MessageStanza stanza_offline : offlines) {
                replies.addReply(channel, stanza_offline);
            }
        }
        return replies;
    }

}
